package com.zhang.sql2;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;

/**
 * @title: 标量函数
 * @author: zhang
 * @date: 2022/3/26 16:38
 * 自定义求hash的函数
 */
public class MyScalarFunction  {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //2 创建表 (连接器表)
        String create = "create table clickTable (" +
                " username STRING," +
                " url STRING," +
                " ts BIGINT" +
                " ) WITH ( " +
                " 'connector' = 'filesystem'," +
                " 'path' = 'input/click.txt'," +
                " 'format' = 'csv'" +
                " )";
        tableEnv.executeSql(create);


        //注册自定义函数
        tableEnv.createTemporarySystemFunction("MyHash",MyHash.class);


        //调用自定义函数
        Table table = tableEnv.sqlQuery("" +
                "select username,MyHash(username) from clickTable");

        //转换成流打印输出
        tableEnv.toDataStream(table).print();
        env.execute();
    }



    public static class MyHash  extends ScalarFunction{
        public int eval(String str){
            return str.hashCode();
        }
    }
}
